Two gaps in coverage worth closing while we're here:

• TTL disabled path. No test exercises the early-return in maybeEnableShortTermMemoryTTL (TTL_MS = 0 or unset). That branch is the default for every existing pipeline, so a regression that accidentally enables TTL with 0ms would be silent. A short scenario with ttlMs = 0 confirming the third event1 returns EXISTING regardless of sleep would cover it.
• Default update-type / visibility. Both tests explicitly override the update-type to OnCreateAndWrite. The option defaults (OnReadAndWrite / NeverReturnExpired) are never actually exercised, so a regression in how the defaults are wired would slip through. One scenario that sets only TTL_MS and asserts the same expiry behavior would catch that.

Nice — recording "Flink does not allow zero/negative TTL" at the workaround site spares future maintainers a trip through Flink source.

Nit: 0L doubles as the "TTL disabled" sentinel, but that contract only lives in OperatorStateManager.maybeEnableShortTermMemoryTTL. A one-line javadoc here — e.g. "Set to a positive value in milliseconds to enable TTL; 0 (the default) disables it" — would spare future readers a trip into the runtime. Same applies to the two enum options below: worth noting they're only consulted when TTL_MS > 0.

Nit: testTTLConfigurationNotApplied vs testTTLConfigurationApplied (line 109) is a bit misleading — TTL is configured in both; what differs is whether enough time elapsed for entries to expire (sleep_ms=0 vs 2000). Names like testValueStillVisibleBeforeTTLExpiry / testValueExpiresAfterTTL would express the actual assertion. Same suggestion for the Python mirror in short_term_memory_ttl_test.py:126,159.

I suggest placing these two methods in UpdateType and StateVisibility, respectively. They are only related to the two enum types themselves, not to where they are used.

hi @xintongsong ,I’d prefer not to put the mapping methods on the public enum types.

Are we comfortable exposing StateTtlConfig.UpdateType and StateTtlConfig.StateVisibility directly in the public API? Two of the new options take Flink internal enums as their value type, in a class under org.apache.flink.agents.api.agents. That couples the public surface to Flink's state API forever, and makes the Python mirror painful (we'd need to redefine the enums on the Python side or fall back to strings).

As mentioned in the earlier review from @weiqingy , exposing Flink’s StateTtlConfig.UpdateType /
StateTtlConfig.StateVisibility directly would couple the Agents public API to Flink state APIs and
make the Python mirror harder. If ShortTermMemoryTtlUpdate or ShortTermMemoryTtlVisibility exposes
toFlink...() methods, we effectively reintroduce that coupling through the enum API.

I agree the mapping does not have to stay in OperatorStateManager.A good compromise would be to move it into a runtime-internal helper, for example:

final class ShortTermMemoryTtlConfigMapper {                                                        
      static StateTtlConfig.UpdateType toFlinkUpdateType(ShortTermMemoryTtlUpdate updateType) {       
          ...                                                                                         
      }                                                                                               
                                                                                                      
      static StateTtlConfig.StateVisibility toFlinkStateVisibility(                                   
              ShortTermMemoryTtlVisibility visibility) {                                              
          ...                                                                                         
      }                                                                                               
  }

Makes sense to me. Thanks for the clarification.

This change is unrelated to TTL. It should be placed in a separate commit. Usually, in such cases, we add [hotfix] commits for fixing existing issues before commits of the actual PR changes.

It's unnecessary to take the whole AgentPlan as an argument. AgentConfiguration should be enough.

Sleeping for 2s is unnecessarily long. If we expect the state still exist after the sleep, sleep shorter (e.g., 50ms). If we expect the state does not exist, we can set a small ttl (e.g., 50ms) and wait a bit longer (e.g., 200ms). Either way, 2s is too long.